package distributedbank;

import java.io.BufferedReader;
import java.io.DataInputStream;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.ObjectOutputStream;
import java.io.Serializable;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

/**
 * The wrapper parses the topology file and takes charge of serializing and
 * sending message objects. The Wrapper will refuse to send messages to
 * dis-allowed JVMs.
 * 
 * 
 * Note that we require all JVM names in the topology file to be of the form
 * "JVM#" where "#" is a digit.
 */
public class Wrapper
{
    /**
     * each JVM will listen on port = (basePort + JVM Number)
     */
    public static final int basePort = 20000;

    /**
     * Map JVM number to a list of JVM Numbers it is able to reach
     */
    private static Map<Integer, List<Integer>> topologyMapping;

    /**
     * static constructor
     */
    static {
        topologyMapping = new HashMap<Integer, List<Integer>>();
        readTopology();
    }

    /**
     * Read topology_file.txt
     */
    private static void readTopology() {
        try {
            FileInputStream fstream = new FileInputStream( "topology_file.txt" );
            DataInputStream in = new DataInputStream( fstream );
            BufferedReader br = new BufferedReader( new InputStreamReader( in ) );
            String strLine;
            while( (strLine = br.readLine()) != null ) {
                updateTopologyMap( Integer.parseInt( String.valueOf( strLine
                        .charAt( 3 ) ) ), Integer.parseInt( String
                        .valueOf( strLine.charAt( 8 ) ) ) );
            }
            in.close();
        } catch( Exception e ) {
            System.err.println( "Error: " + e.getMessage() );
        }
    }

    /**
     * Insert one mapping information into topologyMapping
     */
    private static void updateTopologyMap( int src, int sink ) {
        if( topologyMapping.containsKey( src ) ) {
            topologyMapping.get( src ).add( sink );
        } else {
            List<Integer> list = new ArrayList<Integer>();
            list.add( sink );
            topologyMapping.put( src, list );
        }
    }

    /**
     * Attempts to send a Message to the JVM with the given id. Returns true if
     * the it is permitted to send a message to the given JVM. Otherwise returns
     * false.
     */
    public boolean sendMessage( Serializable message, int srcJvmId, int dstJvmId ) {
        if( !whoNeighbors( srcJvmId ).contains( dstJvmId ) ) {
            return false;
        } else {
            try {
            	Socket socket = new Socket( "localhost", dstJvmId + basePort );
                ObjectOutputStream out = new ObjectOutputStream( socket
                        .getOutputStream() );
                try {
                    out.writeObject( message );
                } finally {
                    out.close();
                    socket.close();
                }
            } catch( IOException e ) {
                System.out.println( e.getMessage() );
            }
            return true;
        }
    }

    /**
     * Look up successors 
     * 
     * @param JVMNumber
     * 		JVM number
     * @return 
     * 		the list of JVM numbers that the given JVMNumber is able to reach directly
     */
    public static List<Integer> whoNeighbors( int JVMNumber ) {
        List<Integer> neighbors = topologyMapping.get( JVMNumber );
        return neighbors == null ? new ArrayList<Integer>() : neighbors;
    }

    /**
     * Look up predecessors 
     * NOTE: ONLY BANK BRANCH PREDESESSORS
     * 
     * @param JVMNumber
     * 		JVM number
     * @return 
     * 		the list of JVM numbers that is able to reach the given JVM directly
     */
    public static List<Integer> incomingChannels( int JVMNumber ) {
        List<Integer> incoming = new ArrayList<Integer>();
        for( Entry<Integer, List<Integer>> pair : topologyMapping.entrySet() ) {
            if( pair.getValue().contains( JVMNumber ) ) {
            	if(pair.getKey() < 5){
            		incoming.add( pair.getKey() );
            	}
            }
        }
        return incoming;
    }
}